package com.dtruth.kafka.consumer;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import kafka.consumer.*;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import org.apache.commons.collections.CollectionUtils;

/**
 * <pre>
 * Created by zhaoming on 14-5-4 下午3:32
 * </pre>
 */
public class kafkaConsumer {

	public static void main(String[] args) throws InterruptedException,
			UnsupportedEncodingException {

		Properties properties = new Properties();
		properties.put("zookeeper.connect", "tree:2181");
		properties.put("auto.commit.enable", "true");
		properties.put("auto.commit.interval.ms", "60000");
		properties.put("group.id", "0");

		ConsumerConfig consumerConfig = new ConsumerConfig(properties);

		ConsumerConnector javaConsumerConnector = Consumer
				.createJavaConsumerConnector(consumerConfig);

		// topic的过滤器
		Whitelist whitelist = new Whitelist("LABEL");
		List<KafkaStream<byte[], byte[]>> partitions = javaConsumerConnector
				.createMessageStreamsByFilter(whitelist);

		if (CollectionUtils.isEmpty(partitions)) {
			System.out.println("empty!");
			TimeUnit.SECONDS.sleep(1);
		}

		// 消费消息
		for (KafkaStream<byte[], byte[]> partition : partitions) {
			ConsumerIterator<byte[], byte[]> iterator = partition.iterator();
			while (iterator.hasNext()) {
				MessageAndMetadata<byte[], byte[]> next = iterator.next();
				System.out.println("partiton:" + next.partition());
				System.out.println("offset:" + next.offset());
				System.out.println("message:"
						+ new String(next.message(), "utf-8"));
			}

		}

	}
}